/*
 *@author  chengkenli
 *@project setbuckets
 *@package tools
 *@file    object
 *@date    2024/7/29 14:38
 */

package tools

import (
    "StarRocksBuckets/conn"
    "StarRocksBuckets/util"
    "encoding/json"
    "errors"
    "fmt"
    "github.com/fatih/color"
    "gorm.io/gorm"
    "strconv"
    "strings"
    "sync"
    "time"
)

type Schemas struct {
    CreateTime string
    Comment    string
    Size       string
    RowCount   string
    SchemaType string
    Nil        string
    Nos        string
    Sum        int
    Enable     string
    Mv         []string
}

func Other() Schemas {
    db, err := conn.StarRocks(util.StarRocks)
    if err != nil {
        fmt.Println(err.Error())
        return Schemas{}
    }

    schema := strings.Split(util.Table, ".")

    //提取创建日期
    var m map[string]interface{}
    sql := fmt.Sprintf("select `TABLE_CATALOG`,`TABLE_SCHEMA`,`TABLE_NAME`,`TABLE_TYPE`,`ENGINE`,`CREATE_TIME`,`TABLE_COMMENT`,`TABLE_ROWS` from information_schema.tables where TABLE_SCHEMA='%s' and TABLE_NAME='%s'", schema[0], schema[1])
    r := db.Raw(sql).Scan(&m)
    if r.Error != nil {
        util.Logger.Error(r.Error.Error())
        return Schemas{}
    }
    //提取表总容量大小和行数
    var s []map[string]interface{}
    r = db.Raw("show data from " + util.Table).Scan(&s)
    if r.Error != nil {
        util.Logger.Error(r.Error.Error())
        return Schemas{}
    }

    //PARTITION BY RANGE
    var cqm []map[string]interface{}
    r = db.Raw(fmt.Sprintf("show create table %s", util.Table)).Scan(&cqm)
    if r.Error != nil {
        util.Logger.Error(err.Error())
        return Schemas{}
    }
    var createSQL string
    for _, m := range cqm {
        createSQL = fmt.Sprintf("%v", m["Create Table"])
    }

    var enable string
    ty := "full"
    for _, s2 := range strings.Split(createSQL, "\n") {
        if strings.Contains(s2, "PARTITION BY RANGE") {
            ty = "incre"
        }
        c := color.New()
        if strings.Contains(s2, "dynamic_partition.enable") {
            a := strings.NewReplacer(`"`, "", " ", "", ",", "").Replace(strings.Split(s2, "=")[1])
            if a == "true" {
                enable = c.Add(color.FgHiGreen).Sprint(a)
            } else if a == "false" {
                enable = c.Add(color.FgHiRed).Sprint(a)
            }
        }
    }
    //采集空分区和非空分区数量
    nil, nos, sum := IsPartition(db)
    mv := Materialized(db, util.Table)

    return Schemas{
        CreateTime: m["CREATE_TIME"].(time.Time).Format("2006-01-02 15:04:05"),
        Comment:    m["TABLE_COMMENT"].(string),
        Size:       s[0]["Size"].(string),
        RowCount:   s[0]["RowCount"].(string),
        SchemaType: ty,
        Nil:        nil,
        Nos:        nos,
        Enable:     enable,
        Sum:        sum,
        Mv:         mv,
    }
}

// SumMapValues 定义一个辅助函数，用于计算map中所有int64值的总和
func SumMapValues(m map[string]int64) int64 {
    sum := int64(0)
    for _, v := range m {
        sum += v
    }
    return sum
}

// RemoveDuplicateStrings /*数组去重*/
func RemoveDuplicateStrings(strs []string) []string {
    result := []string{}
    tempMap := map[string]byte{} // 存放不重复字符串
    for _, e := range strs {
        l := len(tempMap)
        tempMap[e] = 0
        if len(tempMap) != l { // 加入map后，map长度变化，则元素不重复
            result = append(result, e)
        }
    }
    return result
}

// 寻找元素位置
func FindKeyRank(slice []map[string]int64, searchKey string) (int, bool) {
    rank := 0 // 排行计数器
    // 遍历切片中的每个map
    for _, m := range slice {
        // 检查map中是否存在指定的键
        if _, ok := m[searchKey]; ok {
            // 如果找到了匹配的键，返回排行和true
            return rank, true // 排名从1开始计算
        }
        // 增加排行计数器，即使当前map中没有找到键
        rank += len(m)
    }
    // 如果没有找到，返回0和false
    return 0, false
}

// ScanCountDistinct 统计每个字段1000行内重复次数
func ScanCountDistinct(column, table string) int64 {
    db, err := conn.StarRocks(util.StarRocks)
    if err != nil {
        fmt.Println(err.Error())
        return 0
    }
    var m []map[string]interface{}
    sql := fmt.Sprintf(`select count(*),%s from (select DISTINCT %s from (select %s from %s limit 1000) a ) b group by %s`, column, column, column, table, column)
    r := db.Raw(sql).Scan(&m)
    if r.Error != nil {
        util.Logger.Error(r.Error.Error())
        return 0
    }
    return r.RowsAffected
}

// CheckSum 计算两端数据流
func CheckSum(db *gorm.DB, t1, t2 string) (string, int64, int64) {
    fmt.Println("计算两端数据量")
    var c1, c2 map[string]interface{}

    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        db.Raw("select count(*) as count from " + t1).Scan(&c1)
    }()
    go func() {
        defer wg.Done()
        db.Raw("select count(*) as count from " + t2).Scan(&c2)
    }()
    wg.Wait()

    if c1["count"] == nil || c2["count"] == nil {
        return "Fail", -1, -1
    }

    fmt.Println(fmt.Sprintf("[%d]: %s", c1["count"].(int64), t1))
    fmt.Println(fmt.Sprintf("[%d]: %s", c2["count"].(int64), t2))

    if c1["count"].(int64) == c2["count"].(int64) {
        return "Ok", c1["count"].(int64), c2["count"].(int64)
    } else {
        return "Fail", c1["count"].(int64), c2["count"].(int64)
    }
}

// Submit 异步提交
func Submit(sql, taskname, tablename string, db *gorm.DB) error {
    type run struct {
        QUERYID      string `bson:"QUERY_ID"`
        TASKNAME     string `bson:"TASK_NAME"`
        CREATETIME   string `bson:"CREATE_TIME"`
        FINISHTIME   string `bson:"FINISH_TIME"`
        STATE        string `bson:"STATE"`
        DATABASE     string `bson:"DATABASE"`
        DEFINITION   string `bson:"DEFINITION"`
        EXPIRETIME   string `bson:"EXPIRE_TIME"`
        ERRORCODE    string `bson:"ERROR_CODE"`
        ERRORMESSAGE string `bson:"ERROR_MESSAGE"`
        PROGRESS     string `bson:"PROGRESS"`
        EXTRAMESSAGE string `bson:"EXTRA_MESSAGE"`
    }
    r := db.Exec(sql)
    if r.Error != nil {
        util.Logger.Error(r.Error.Error())
        return r.Error
    }

    var i int
    ticker := time.NewTicker(time.Second * 1)
    for {
        select {
        case <-ticker.C:
            //c := color.New()
            s := fmt.Sprintf("select * from information_schema.task_runs where task_name ='%s'", taskname)
            var task run
            r := db.Raw(s).Scan(&task)
            if r.Error != nil {
                util.Logger.Error(r.Error.Error())
                return r.Error
            }

            if i%60 == 0 {
                fmt.Println(fmt.Sprintf("%s  %s  %s  %s", time.Now().Format("2006-01-02 15:04:05"), task.STATE, task.PROGRESS, tablename))
            }

            if s == "" {
                return errors.New("information_schema.task_runs is nil")
            }
            if task.STATE == "SUCCESS" {
                return nil
            }
            if task.STATE == "FAILED" {
                return errors.New("task staus is " + task.STATE)
            }
            i++
        }
    }
}

/*匹配starrocks版本*/
func Versions(db *gorm.DB) float64 {
    type Version struct {
        Version string `bson:"version"`
    }
    var v Version
    sqln := fmt.Sprintf("select current_version() as version")
    db.Raw(sqln).Scan(&v)
    version, _ := strconv.ParseFloat(fmt.Sprintf("%s.%s", strings.Split(strings.Split(v.Version, " ")[0], ".")[0], strings.Split(strings.Split(v.Version, " ")[0], ".")[1]), 64)
    return version
}

func Materialized(db *gorm.DB, table string) []string {
    var m map[string]interface{}
    r := db.Raw(fmt.Sprintf("select inspect_related_mv('%s') as mv", table)).Scan(&m)
    if r.Error != nil {
        fmt.Println(r.Error.Error())
        return nil
    }
    if m == nil || m["mv"] == nil {
        fmt.Println("inspect_related_mv is nil")
        return nil
    }
    fmt.Println(m["mv"].(string))

    type Data []struct {
        ID   int    `json:"id"`
        Name string `json:"name"`
    }
    var d Data
    err := json.Unmarshal([]byte(m["mv"].(string)), &d)
    if err != nil {
        fmt.Println(err.Error())
        return nil
    }

    c := color.New()
    var mv []string
    for i, s := range d {
        mv = append(mv, fmt.Sprintf("%s>%s", c.Add(color.FgHiGreen).Sprint(i), c.Add(color.FgHiMagenta).Sprint(s.Name)))
    }
    return mv
}
